package org.seed.quantum.message.rpc;

import org.seed.quantum.message.plugins.Plugin;
import org.seed.quantum.socket.AbstractSocketSession;
import org.seed.quantum.socket.MessageProcessor;
import org.seed.quantum.socket.NetMonitor;
import org.seed.quantum.socket.SocketStateEnum;

import java.io.*;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

public class RpcProviderProcessor implements MessageProcessor<byte[]>, NetMonitor {

    private Map<String, Object> impMap = new HashMap<String, Object>();
    private ExecutorService pool = Executors.newCachedThreadPool();
    /**
     * 基础数据类型
     */
    private Map<String, Class<?>> primitiveClass = new HashMap<String, Class<?>>();

    // 9种基本数据类型
    {
        primitiveClass.put("boolean", boolean.class);
        primitiveClass.put("byte", byte.class);
        primitiveClass.put("char", char.class);
        primitiveClass.put("double", double.class);
        primitiveClass.put("float", float.class);
        primitiveClass.put("int", int.class);
        primitiveClass.put("long", long.class);
        primitiveClass.put("short", short.class);
        primitiveClass.put("void", void.class);
    }

    public RpcProviderProcessor() {
    }

    private Consumer<SocketStateEnum> sessionStateConsumer;

    public RpcProviderProcessor(Consumer<SocketStateEnum> stateConsumer) {
        this.sessionStateConsumer = stateConsumer;
    }

    @Override
    public void process(AbstractSocketSession session, byte[] msg) {
        pool.execute(() -> {
            ObjectInput objectInput = null;
            ObjectOutput objectOutput = null;
            try {
                objectInput = new ObjectInputStream(new ByteArrayInputStream(msg));
                RpcRequest req = (RpcRequest) objectInput.readObject();

                RpcResponse resp = new RpcResponse(req.getUuid());
                try {
                    String[] paramClassList = req.getParamClassList();
                    Object[] paramObjList = req.getParams();
                    // 获取入参类型
                    Class<?>[] classArray = null;
                    if (paramClassList != null) {
                        classArray = new Class[paramClassList.length];
                        for (int i = 0; i < classArray.length; i++) {
                            Class<?> clazz = primitiveClass.get(paramClassList[i]);
                            if (clazz == null) {
                                classArray[i] = Class.forName(paramClassList[i]);
                            } else {
                                classArray[i] = clazz;
                            }
                        }
                    }
                    // 调用接口
                    Object impObj = impMap.get(req.getInterfaceClass());
                    if (impObj == null) {
                        throw new UnsupportedOperationException("can not find interface: " + req.getInterfaceClass());
                    }
                    Method method = impObj.getClass().getMethod(req.getMethod(), classArray);
                    Object obj = method.invoke(impObj, paramObjList);
                    resp.setReturnObject(obj);
                    resp.setReturnType(method.getReturnType().getName());
                } catch (InvocationTargetException e) {
                    resp.setException(e);
                } catch (Exception e) {
                    resp.setException(e);
                }
                ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
                objectOutput = new ObjectOutputStream(byteArrayOutputStream);
                objectOutput.writeObject(resp);
                byte[] data = byteArrayOutputStream.toByteArray();
                synchronized (session) {
                    session.writeBuffer().writeInt(data.length + 4);
                    session.writeBuffer().write(data);
                    session.writeBuffer().flush();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            } finally {
                if (objectInput != null) {
                    try {
                        objectInput.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (objectOutput != null) {
                    try {
                        objectOutput.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

    }


    public final <T> void publishService(Class<T> apiName, T apiImpl) {
        impMap.put(apiName.getName(), apiImpl);
    }

    private final List<Plugin<byte[]>> plugins = new ArrayList<>();

    @Override
    public void stateEvent(AbstractSocketSession session, SocketStateEnum socketStateEnum, Throwable throwable) {
        if (socketStateEnum == SocketStateEnum.DECODE_EXCEPTION || socketStateEnum == SocketStateEnum.PROCESS_EXCEPTION) {
            throwable.printStackTrace();
        }
        switch (socketStateEnum) {
            case NEW_SESSION:
            case SESSION_CLOSED:
                if (sessionStateConsumer != null) {
                    sessionStateConsumer.accept(socketStateEnum);
                }
                break;
        }
    }

    @Override
    public final void afterRead(AbstractSocketSession session, int readSize) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.afterRead(session, readSize);
        }
    }

    @Override
    public final void afterWrite(AbstractSocketSession session, int writeSize) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.afterWrite(session, writeSize);
        }
    }

    @Override
    public final void beforeRead(AbstractSocketSession session) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.beforeRead(session);
        }
    }

    @Override
    public final void beforeWrite(AbstractSocketSession session) {
        for (Plugin<byte[]> plugin : plugins) {
            plugin.beforeWrite(session);
        }
    }

    @Override
    public final AsynchronousSocketChannel shouldAccept(AsynchronousSocketChannel channel) {
        AsynchronousSocketChannel acceptChannel = channel;
        for (Plugin<byte[]> plugin : plugins) {
            acceptChannel = plugin.shouldAccept(acceptChannel);
            if (acceptChannel == null) {
                return null;
            }
        }
        return acceptChannel;
    }

    public final void addPlugin(Plugin<byte[]> plugin) {
        this.plugins.add(plugin);
    }
}
